package com.atguigu.day05;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.stream.Collectors;


/**
 * 浏览量排行只开一次窗（性能不好版
 */
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env
                .readTextFile("D:\\WebSite\\flink\\flink\\java_flink\\src\\main\\resources\\UserBehavior.csv")
                .map(x -> {
                    String[] arr = x.split(",");
                    return new UserBehavior(
                            arr[0],
                            arr[1],
                            arr[2],
                            arr[3],
                            Long.parseLong(arr[4])
                    );
                })
                .filter(x -> x.behavior.equals("pv"))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                                    @Override
                                    public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                )
                .keyBy(x -> true)
                .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5)))
                .process(new ProcessWindowFunction<UserBehavior, String, Boolean, TimeWindow>() {
                    @Override
                    public void process(Boolean aBoolean, Context context, Iterable<UserBehavior> elements, Collector<String> out) throws Exception {
                        //将窗口中的每一个itemId和它出现的次数保存在hashmap中
                        HashMap<String, Long> map = new HashMap<>();
                        for (UserBehavior ub : elements) {
                            if (map.containsKey(ub.itemId)){
                                map.put(ub.itemId, map.get(ub.itemId)+1L);
                            }else {
                                map.put(ub.itemId, 1L);
                            }
                        }
                        //将hashmap的遍历保存在list集合中
                        ArrayList<ItemViewCount> list = new ArrayList<>();
                        map.keySet().forEach(ub->{
                            list.add(new ItemViewCount(
                                    ub,
                                    context.window().getStart(),
                                    context.window().getEnd(),
                                    map.get(ub)
                            ));
                        });
                        //对list进行排序
                        List<ItemViewCount> collect = list.stream().sorted(new Comparator<ItemViewCount>() {
                            @Override
                            public int compare(ItemViewCount o1, ItemViewCount o2) {
                                return (int) (o2.count-o1.count);
                            }
                        }).collect(Collectors.toList());
                        //组成打印结果
                        StringBuilder sb = new StringBuilder();
                        sb.append("****************************\n");
                        for (int i = 0; i < 3; i++) {
                            sb.append("窗口结束时间：")
                                    .append(new Timestamp(context.window().getEnd()))
                                    .append("第").append(i + 1).append("名的商品ID是：")
                                    .append(collect.get(i).itemId).append(", 浏览量是：")
                                    .append(collect.get(i).count)
                                    .append("\n");
                        }
                        sb.append("****************************\n");
                        out.collect(sb.toString());
                    }
                })
                .print();
        env.execute();
    }

    public static class ItemViewCount {
        public String itemId;
        public Long startTime;
        public Long endTime;
        public Long count;

        public ItemViewCount() {
        }

        @Override
        public String toString() {
            return "ItemViewCount{" +
                    "itemId='" + itemId + '\'' +
                    ", startTime=" + new Timestamp(startTime) +
                    ", endTime=" + new Timestamp(endTime) +
                    ", count=" + count +
                    '}';
        }

        public ItemViewCount(String itemId, Long startTime, Long endTime, Long count) {
            this.itemId = itemId;
            this.startTime = startTime;
            this.endTime = endTime;
            this.count = count;
        }
    }


    public static class UserBehavior {
        public String userId;
        public String itemId;
        public String categoryId;
        public String behavior;
        public Long timestamp;

        @Override
        public String toString() {
            return "UserBehavior{" +
                    "userId='" + userId + '\'' +
                    ", itemId='" + itemId + '\'' +
                    ", categoryId='" + categoryId + '\'' +
                    ", behavior='" + behavior + '\'' +
                    ", timestamp=" + new Timestamp(timestamp) +
                    '}';
        }

        public UserBehavior() {
        }

        public UserBehavior(String userId, String itemId, String categoryId, String behavior, Long timestamp) {
            this.userId = userId;
            this.itemId = itemId;
            this.categoryId = categoryId;
            this.behavior = behavior;
            this.timestamp = timestamp * 1000L;
        }
    }

}
